package com.eastcom.collection.distributor


import org.apache.log4j.Logger;
import java.util.ArrayList;






import com.eastcom.collection.tools.db.DbEngine
import java.sql.Date;


import java.sql.Timestamp
import java.sql.Types;
import java.text.SimpleDateFormat
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.Socket;
import java.util.ArrayList;

import com.eastcom.collection.provider.*



/**
 * Processor的接收线程。 负责接收客户端发来的数据并转送到处理线程中。
 *
 * @author 张海峰
 * @version 1.0
 */
public class WorkerThread extends Thread {
	//~ Static fields/initializers ---------------------------------------------

	private static Logger logger = Logger.getLogger(WorkerThread.class);
	
	private static List<String> cache = new ArrayList<String>()
	//~ Instance fields --------------------------------------------------------
	//private int serNum;
	private Socket       socket = null;

	//~ Constructors -----------------------------------------------------------

	/**
	 * Creates a new WorkerThread object.
	 *
	 * @param socket DOCUMENT ME!
	 * @param cfi DOCUMENT ME!
	 * @param tt DOCUMENT ME!
	 */
	public WorkerThread(Socket socket) {
		this.socket = socket;
		//serNum=createSerNum();
		//logger.info("Socket thread-"+this.serNum+" was established.");
		this.start();
	}

	//~ Methods ----------------------------------------------------------------

	/**
	 * DOCUMENT ME!
	 */
	public void run() {
		println "work thread starting..."
		long time = new Date().getTime()

		
		InputStream inp=null
		try {
			//InputStream in = new BufferedInputStream(socket.getInputStream());
			inp = new BufferedInputStream(socket.getInputStream());
			//byte[] ba = new
			// byte[SimpleKit.toInt(cfi.getValue("MaxLineSize"))];
			byte[] baLen  = new byte[QueueConst.LENGTH_FLAG_LENGTH];
			byte[] ba;
			int    sepLen = QueueConst.WORD_SEPARATOR.length();

			while (true) {
				try {
					int count0 = inp.read(baLen);

					if (count0 == -1) {
						break;
					}

					while (count0 < QueueConst.LENGTH_FLAG_LENGTH) {
						count0 += inp.read(baLen, count0,
								QueueConst.LENGTH_FLAG_LENGTH -
								count0);
					}

					int i = Integer.valueOf(new String(baLen)) -
							QueueConst.LENGTH_FLAG_OFFSET;

					if (i < 0) {
						logger.error("Invalid number，<0:" + i);

						break;
					}

					try {
						ba = new byte[i];
					} catch (OutOfMemoryError oom) {
						//                        tt.clearBank();
						ba = new byte[i];
						logger.error("Memory exhausted, clear buffer. ");
					}

					int counts = inp.read(ba);

					while (counts < i) {
						counts += inp.read(ba, counts, i - counts);
					}
					cache.add(new String(ba))
					//					if(parseLine(new String(ba))==null){
					//						continue
					//					}
					logger.info("" + ++counts);
				} catch (OutOfMemoryError oom) {
					logger.error("Memory exhausted, clear buffer.");
				}
			}
		} catch (Exception e) {
			//e.printStackTrace();
			logger.error(e);
		}finally{
			if(inp!=null)
			{
				try{
					inp.close();

				}catch(IOException ioe)
				{}
				inp=null;
			}
			if(this.socket!=null)
			{
				try{
					this.socket.close();
				}catch(IOException ioe)
				{
					logger.error(ioe);
				}
				this.socket=null;
			}
			//logger.info("Socket thread-"+this.serNum+" was closed.");
		}
		for(String line: cache){
			DPService.cacheData.add(line)
		}
		cache.clear()
	}
}